package com.example.demo.demos.utils;

import com.example.demo.demos.mapper.OrderMapper;
import com.example.demo.demos.pojo.Order;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

@Component
public class OrderKafkaComsumer {
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private RedisTemplate redisTemplate;

    //监听消费
    @KafkaListener(topics = {"topicname"})
    public void onNormalMessage(ConsumerRecord<String, Object> record) {
        String[] split = record.value().toString().split(",");
        if (split.length>=3) {
            Order order = new Order();
            order.setOrderId(split[0])
                    .setOrderdata(split[1])
                    .setMoney(Float.parseFloat(split[2]));
            Integer i = orderMapper.addOrder(order);
            OrderInfo orderInfo = orderMapper.queryStatus(split[0]);
            if (i>0 && orderInfo!=null){
                redisTemplate.opsForValue().set("orderId:"+ orderInfo.getOrderId(),orderInfo.toString());
            }
        }
    }

}
